package com.dyna.web.websocket;

import com.dyna.core.utils.ObjectUtil;
import lombok.Data;
import lombok.extern.slf4j.Slf4j;

import javax.websocket.*;
import javax.websocket.server.PathParam;
import java.io.IOException;
import java.util.List;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.CopyOnWriteArraySet;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * 前台消息推送ws
 * WebSocketServer：每个连接各自的服务端对象 每连接一个对象，spring就自动新建一个WebSocketServer对象
 * WebSocketServerSet：所有WebSocket对象集合
 * userId：每个WebSocket对象的用户id标识
 *
 * @author zhouqing
 * createTime 2021/10/22
 * lastModify zhouqing
 * lastModifyTime 2021/10/22
 * 创新中心
 */
@Slf4j
//@ServerEndpoint("/webSocket/{userId}")
//@Component
@Data
public class WebSocketServer {

    /**
     * 静态变量，用来记录当前在线连接数。应该把它设计成线程安全的
     */
    private static AtomicInteger onlineCount = new AtomicInteger(0);
    /**
     * concurrent包的线程安全Set，用来存放每个客户端对应的MyWebSocket对象。
     * 因为spring容器默认是单例模式，每次产生新连接，生成新的WebSocketServer对象，前一个连接对象就被覆盖了，所以需要保存到这个集合中
     */
    private static CopyOnWriteArraySet<WebSocketServer> webSocketSet = new CopyOnWriteArraySet<>();
    private static final ConcurrentHashMap<Long, Session> ONLINE_WS_MAP = new ConcurrentHashMap<>();

    /**
     * 接收userId
     */
    private Long userId;

    /**
     * 连接建立成功调用的方法
     * userId接收前端传过来的参数
     */
    @OnOpen
    public void onOpen(Session session, @PathParam("userId") Long userId) {
        if (ObjectUtil.isNull(ONLINE_WS_MAP.get(userId))) {
            ONLINE_WS_MAP.put(userId, session);
            //在线数加1
            onlineCount.incrementAndGet();
            log.info("SysNoticeWebSocketServer有新用户接入:" + userId + ",当前在线用户数为" + ONLINE_WS_MAP.size());
            this.userId = userId;
        }else {
            log.error("用户【{}】已连接......", userId);
        }
    }

    /**
     * 连接关闭调用的方法
     */
    @OnClose
    public void onClose() {
        if (ObjectUtil.isNotNull(ONLINE_WS_MAP.get(userId))) {
            //从set中删除
            ONLINE_WS_MAP.remove(this.userId);
            //在线数减1
            onlineCount.decrementAndGet();
            log.info("SysNoticeWebSocketServer有一用户连接关闭！当前在线用户数为" + onlineCount.get());
        }
    }


    /**
     * @param message json字符串
     * @param session session
     */
    @OnMessage
    public void onMessage(String message, Session session) {
        log.info("收到来自用户【{}】的消息:{}", userId, message);
    }


    /**
     * @param session session
     * @param error error
     */
    @OnError
    public void onError(Session session, Throwable error) {
        log.error("websocket error info ::: {}", error.getMessage());
        if (ObjectUtil.isNotNull(ONLINE_WS_MAP.get(userId))) {
            //从set中删除
            ONLINE_WS_MAP.remove(userId);
            //在线数减1
            onlineCount.decrementAndGet();
            log.info("SysNoticeWebSocketServer有一用户连接关闭！当前在线用户数为" + onlineCount.get());
        }
    }


    /**
     * 实现服务器主动推送
     */
    private void sendMessage(String message) throws IOException {
        Session session = ONLINE_WS_MAP.get(userId);
        if (ObjectUtil.isNotNull(session)) {
            session.getBasicRemote().sendText(message);
        }

    }


    /**
     * 群发自定义消息
     */
    public void sendInfo(String message, Integer userId) {
        ONLINE_WS_MAP.forEach((k,v)-> {
            //这里可以设定只推送给这个userId的，为null则全部推送
            if (userId == null) {
                this.sendMsg(message, userId, v);
            }
        });
    }

    /**
     * 群发自定义消息
     */
    public void sendInfo(String message, List<Integer> uidList) {
        uidList.forEach(uid -> sendInfo(message, uid));
    }


    private void sendMsg(String message, Integer uid, Session session) {
        //防止给断开的连接，发数据，报错导致阻塞
        if (ObjectUtil.isNotNull(session) && session.isOpen()) {
            try {
                sendMessage(message);
            } catch (IOException e) {
                log.error("向id为" + uid + "的用户消息推送" + message + "失败", e);
            }
        }
    }

}